package com.bff.gaia.api.common.functions;

import com.bff.gaia.api.java.tuple.Tuple;
import com.bff.gaia.util.Collector;

public final class ProjectFlatJoinFunction<T1, T2, R extends Tuple> extends RichFlatJoinFunction<T1, T2, R> {

	private static final long serialVersionUID = 1L;

	private final int[] fields;
	private final boolean[] isFromFirst;
	private final R outTuple;

	/**
	 * Instantiates and configures a ProjectJoinFunction.
	 * Creates output tuples by copying fields of joined input tuples (or a full input object) into an output tuple.
	 *
	 * @param fields List of indexes fields that should be copied to the output tuple.
	 * 					If the full input object should be copied (for example in case of a non-tuple input) the index should be -1.
	 * @param isFromFirst List of flags indicating whether the field should be copied from the first (true) or the second (false) input.
	 * @param outTupleInstance An instance of an output tuple.
	 */
	public ProjectFlatJoinFunction(int[] fields, boolean[] isFromFirst, R outTupleInstance) {
		if (fields.length != isFromFirst.length) {
			throw new IllegalArgumentException("Fields and isFromFirst arrays must have same length!");
		}

		this.fields = fields;
		this.isFromFirst = isFromFirst;
		this.outTuple = outTupleInstance;
	}

	public int[] getFields() {
		return fields;
	}

	public boolean[] getIsFromFirst() {
		return isFromFirst;
	}

	public void join(T1 in1, T2 in2, Collector<R> out) {
		for (int i = 0; i < fields.length; i++) {
			if (isFromFirst[i]) {
				if (fields[i] >= 0 && in1 != null) {
					outTuple.setField(((Tuple) in1).getField(fields[i]), i);
				} else {
					outTuple.setField(in1, i);
				}
			} else {
				if (fields[i] >= 0 && in2 != null) {
					outTuple.setField(((Tuple) in2).getField(fields[i]), i);
				} else {
					outTuple.setField(in2, i);
				}
			}
		}
		out.collect(outTuple);
	}

}